use futures::executor::block_on;
use futures::{SinkExt, StreamExt};
use async_std::task;
use std::time::Duration;
use futures::channel::mpsc;

#[test]
fn test_async_channel(){
    // mpsc::unbounded 纯异步队列   ≈   mpsc::channel(max)
    // mpsc::channel  带缓存的同步队列   mpsc::channel(0-max) 跟go的channel类似，为0则阻塞

    let (tx, mut tr) = mpsc::channel::<i32>(0);

    let mut thread_tx = tx.clone();

    task::spawn(async move{
        let mut i : i32 = 0;
        while let Some(res) = tr.next().await {
            i+=1;
            println!("res = {}", res);
            task::sleep(Duration::from_secs(1)).await;
            if i == 3 {
                // 强行关闭channel
                thread_tx.close_channel();
            }
        }
        println!("Receiver found channel closed.");
    });

    let mut thread_tx = tx.clone();

    task::spawn(async move {
        for i in 1..1+5 {
            println!("send {} start", i);
            if let Err(_) = thread_tx.send(i).await {
                println!("Sender found channel closed.");
                break
            }
            println!("send {} end", i);
        }
    });

    let mut thread_tx = tx.clone();

    task::spawn(async move {
        for i in 5..5+5 {
            println!("send {} start", i);
            if let Err(_) = thread_tx.send(i).await {
                println!("Sender found channel closed.");
                break
            }
            println!("send {} end", i);
        }
    });



    block_on(task::sleep(Duration::from_secs(60)));
}
